package indi.fritz.bigmap.volme;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.channels.FileLock;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import indi.fritz.bigmap.common.DBException;
import indi.fritz.bigmap.common.LimitQueue;
import indi.fritz.bigmap.common.Options;
import indi.fritz.bigmap.common.Page;
import indi.fritz.bigmap.volme.data.DataEngine;
import indi.fritz.bigmap.volme.indexing.RBTree;
import indi.fritz.bigmap.volme.model.DataOffset;
import indi.fritz.bigmap.volme.model.KeyValue;
import indi.fritz.bigmap.volme.offset.OffsetEngine;

public class DiskEngine implements Engine{
	ExecutorService pool = new ThreadPoolExecutor(1, 200, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
	
	DataEngine dataEngine;
	OffsetEngine offsetEngine;
	
	RBTree tree;
	
	RandomAccessFile LockFile;
	FileLock lock;
	boolean readonly = false;
	
	public DiskEngine(Options options) throws IOException {
		File file = options.getFolder();
		File dataFile = new File(file, "data");
		File offsetFile = new File(file, "offset");
		/**
		 * 初始化文件
		 */
		if(options.isCreateIfNotExist()){
			file.mkdirs();
			if(!dataFile.exists()){ dataFile.createNewFile(); }
			if(!offsetFile.exists()){ offsetFile.createNewFile(); }
		}
		/**
		 * 写一个标志文件 LOCK 如果这个文件不存在,或者加锁成功,则说明此时没有程序读取当前数据库
		 */
		if(options.isExclusive()){
			File lockFile = new File(file, "LOCK");
			if(!lockFile.exists()){
				lockFile.createNewFile();
			}
			lockFile.deleteOnExit();
			try {
				this.LockFile = new RandomAccessFile(lockFile, "rw");
			    this.lock = this.LockFile.getChannel().tryLock();
			    if(this.lock == null || !this.lock.isValid()){
			    	throw new DBException.ExclusiveVoid();
			    }
			} catch (Exception e) {
				throw new DBException.ExclusiveVoid();
			}
		}
		this.dataEngine = new DataEngine(dataFile);
		this.offsetEngine = new OffsetEngine(offsetFile);
		
		if(options.isClearOnOpen()){ clear(); }
		if(options.isReadonly()){//只读的操作
			this.readonly = true;
		}
		this.init();
	}
	
	private void init() throws IOException {
		RBTree tree = new RBTree();
		for (long i = 0; i < this.count(); i++) {
			KeyValue kv = this.read(i);
			if(kv != null) {
				tree.insert(kv.getKey(), i);
			}
		}
		this.tree = tree;
	}
	
	@Override
	public boolean exist(byte[] key) {
		return this.tree.exist(key);
	}
	
	@Override
	public byte[] read(byte[] key){
		try {
			KeyValue kv = this.readKeyValue(key);
			if(kv == null) { return null; }
			return kv.getValue();
		} catch (Exception e) {
			throw new DBException.DBReadException(e);
		}
	}
	
	public Page<KeyValue> read(long cursor, int pageSize) {
		try {
			long count = this.length();
			List<KeyValue> list = new ArrayList<KeyValue>();
			while(list.size() < pageSize){
				cursor++;
				KeyValue value = this.read(cursor);
				if(value != null){
					list.add(value);
				}
			}
			return new Page<KeyValue>(cursor, list, pageSize, count);
		} catch (Exception e) {
			throw new DBException.DBReadException(e);
		}
	}
	
	@Override
	public long write(byte[] key, byte[] value) {
		if(this.readonly) { throw new DBException.ReadOnlyException();}
		try {
			return pool.submit(new Callable<Long>() {
				@Override
				public Long call() throws Exception {
					Long oldIndex = tree.get(key);
					if(oldIndex == null) {
						DataOffset offset = dataEngine.writeKeyValue(new KeyValue(key, value, true));
						long index = offsetEngine.writeOffset(offset);
						tree.insert(key, index);
						return index;
					}else {
						DataOffset offset = offsetEngine.readOffset(oldIndex);
						offset = dataEngine.updateKeyValue(offset, new KeyValue(key, value, true));
						return offsetEngine.updateOffset(oldIndex, offset);
					}
				}
			}).get();
		} catch (Exception e) {
			throw new DBException.DBWriteException(e);
		}
	}
	
	@Override
	public byte[] delete(byte[] key) {
		if(this.readonly) { throw new DBException.ReadOnlyException();}
		try {
			return pool.submit(new Callable<byte[]>() {
				@Override
				public byte[] call() throws Exception {
					Long index = tree.delete(key);
					if(index == null) { return null; }
					DataOffset offset = offsetEngine.deleteOffset(index);
					KeyValue kv = dataEngine.delete(offset);
					return kv.getValue();
				}
			}).get();
		} catch (Exception e) {
			throw new DBException.DBDeleteException(e);
		}
	}
	
	@Override
	public long length() {
		return this.tree.length();
	}
	
	@Override
	public void forEach(Consumer<KeyValue> action) {
		this.tree.forEach(key->{
			try {
				action.accept(this.readKeyValue(key));
			} catch (IOException e) {
				throw new DBException("read data error" , e);
			}
		});
	}
	
	@Override
	public void clear() {
		if(this.readonly) { throw new DBException.ReadOnlyException();}
		try {
			pool.submit(new Callable<Void>() {
				@Override
				public Void call() throws Exception {
					offsetEngine.clear();
					dataEngine.clear();
					if(tree != null) {
						tree.clear();
					}
					return null;
				}
			}).get();
		} catch (Exception e) {
			throw new DBException.DBDeleteException(e);
		}
	}
	
	@Override
	public void close() {
		try {
			if(this.lock != null){
				this.lock.release();
				this.lock.close();
				this.LockFile.close();
			}
			this.dataEngine.close();
			this.offsetEngine.close();
			pool.shutdown();
		} catch (Exception e) {
			throw new DBException("can not close engine;", e);
		}
	}
	
	private int count() throws IOException {
		return this.offsetEngine.count();
	}
	
	private KeyValue read(long index) throws IOException{
		DataOffset offset = this.offsetEngine.readOffset(index);
		if(offset == null || !offset.isExist()) { return null; }
		KeyValue kv = this.dataEngine.readKeyValue(offset);
		if(kv == null || !kv.isExist()) { return null; }
		return kv;
	}
	
	private KeyValue readKeyValue(byte[] key) throws IOException{
		Long index = this.tree.get(key);
		if(index == null) { return null; }
		DataOffset offset = this.offsetEngine.readOffset(index);
		if(offset == null || !offset.isExist()) { return null; }
		KeyValue kv = this.dataEngine.readKeyValue(offset);
		if(kv == null || !kv.isExist()) { return null; }
		return kv;
	}
	
	public static void rebuildByData(File folder, File rebuildFile) throws IOException{
		File dataFile = new File(folder, "data");
		File offsetFile = new File(folder, "offset");
		if(!folder.exists()) {  throw new DBException("can not rebuild this db; folder not exist;");}
		if(!rebuildFile.exists()) { throw new DBException("can not rebuild this db; because the data file not exist;");}
		if(dataFile.exists() || offsetFile.exists()) {  throw new DBException("can not rebuild this db; this folder is not empty;");}
		try {
			Options options = new Options();
			options.setCreateIfNotExist(true);
			options.setFolder(folder);
			options.setExclusive(true);
			options.setClearOnOpen(true);
			try(DiskEngine engine = new DiskEngine(options)){
				DiskEngine.findKeyValue(rebuildFile, new Consumer<KeyValue>(){
					@Override
					public void accept(KeyValue kv) {
						engine.write(kv.getKey(), kv.getValue());
					}
				});
			}
		} catch (Exception e) {
			if(dataFile.exists()) { dataFile.delete(); }
			if(offsetFile.exists()) { offsetFile.delete(); }
			throw new DBException("can not rebuild this db;", e);
		}
	}
	
	static void findDataOffset(File dataFile, BiConsumer<KeyValue, DataOffset> consumer) throws IOException {
		if(!dataFile.exists()) { throw new DBException("can not find data file"); }
		Byte[] EOFbyte = new Byte[DataEngine.EOF.length];
		for (int i = 0; i < DataEngine.EOF.length; i++) {
			EOFbyte[i] = DataEngine.EOF[i];
		}
		try(RandomAccessFile reader = new RandomAccessFile(dataFile, "r")){
			long seek = EOFbyte.length;
			long length = reader.length();
			LimitQueue<Byte> queue = new LimitQueue<Byte>(EOFbyte.length);
			for (int i = 0; i < EOFbyte.length; i++) {
				reader.seek(i);
				byte b = reader.readByte();
				queue.offer(b);
			}
			long start = 0;
			long end = 0;
			while(seek < length) {
				reader.seek(seek);
				seek++ ;
				if(seek == length) { return; }
				byte b = reader.readByte();
				queue.offer(b);
				boolean flag = Arrays.equals(queue.toArray(new Byte[EOFbyte.length]), EOFbyte);
				if(flag) {
					end = seek;
					byte[] dataByte = new byte[(int) (end - start)];
					reader.read(dataByte, 0, dataByte.length);
					KeyValue kv = KeyValue.format(dataByte);
					consumer.accept(kv, new DataOffset(start, (int) (end - start), true));
					start = end;
				}
			}
		}
	}
	
	static void findKeyValue(File dataFile, Consumer<KeyValue> consumer) throws IOException {
		if(!dataFile.exists()) { throw new DBException("can not find data file"); }
		Byte[] EOFbyte = new Byte[DataEngine.EOF.length];
		for (int i = 0; i < DataEngine.EOF.length; i++) {
			EOFbyte[i] = DataEngine.EOF[i];
		}
		try(RandomAccessFile reader = new RandomAccessFile(dataFile, "r")){
			long seek = EOFbyte.length;
			long length = reader.length();
			LimitQueue<Byte> queue = new LimitQueue<Byte>(EOFbyte.length);
			for (int i = 0; i < EOFbyte.length; i++) {
				reader.seek(i);
				byte b = reader.readByte();
				queue.offer(b);
			}
			long start = 0;
			long end = 0;
			while(seek < length) {
				reader.seek(seek);
				seek++ ;
				if(seek == length) { return; }
				byte b = reader.readByte();
				queue.offer(b);
				boolean flag = Arrays.equals(queue.toArray(new Byte[EOFbyte.length]), EOFbyte);
				if(flag) {
					end = seek;
					byte[] dataByte = new byte[(int) (end - start)];
					reader.read(dataByte);
					KeyValue kv = KeyValue.format(dataByte);
					if(kv.isExist()) { consumer.accept(kv); }
					start = end;
				}
			}
		}
	}
}
